package sql

import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}

object sql02_wc_tb {

  def main(args: Array[String]): Unit = {


    val ss: SparkSession = SparkSession.builder()
      .master("local[*]")
      .appName("sql02")
      .getOrCreate()


    ss.sparkContext.setLogLevel("TRACE")


    import ss.implicits._

    val df: DataFrame = ss.createDataFrame(List(
      (1, "aa", 100),
      (2, "bb", 200),
      (3, "cc", 300),
      (4, "dd", 400),
      (5, "ee", 500)
    )).toDF("id", "name", "score")



    df.createTempView("t_user")


    val res1: DataFrame = ss.sql("select ta.id,ta.name,tb.score " +
      " from " +
      " (select id, name from t_user) as ta " +
      " join " +
      " (select id, score from t_user) as tb " +
      " on " +
      " ta.id = tb.id " +
      " where tb.score > 200 ")

 /**

    == Parsed Logical Plan ==
'Project ['ta.id, 'ta.name, 'tb.score]
+- 'Filter ('tb.score > 200)
   +- 'Join Inner, ('ta.id = 'tb.id)
      :- 'SubqueryAlias ta
      :  +- 'Project ['id, 'name]
      :     +- 'UnresolvedRelation `t_user`
      +- 'SubqueryAlias tb
         +- 'Project ['id, 'score]
            +- 'UnresolvedRelation `t_user`

== Analyzed Logical Plan ==
id: int, name: string, score: int
Project [id#6, name#7, score#32]
+- Filter (score#32 > 200)
   +- Join Inner, (id#6 = id#30)
      :- SubqueryAlias ta
      :  +- Project [id#6, name#7]
      :     +- SubqueryAlias t_user
      :        +- Project [_1#0 AS id#6, _2#1 AS name#7, _3#2 AS score#8]
      :           +- LocalRelation [_1#0, _2#1, _3#2]
      +- SubqueryAlias tb
         +- Project [id#30, score#32]
            +- SubqueryAlias t_user
               +- Project [_1#0 AS id#30, _2#1 AS name#31, _3#2 AS score#32]
                  +- LocalRelation [_1#0, _2#1, _3#2]

== Optimized Logical Plan ==
Project [id#6, name#7, score#32]
+- Join Inner, (id#6 = id#30)
   :- LocalRelation [id#6, name#7]
   +- Project [_1#0 AS id#30, _3#2 AS score#32]
      +- Filter (_3#2 > 200)
         +- LocalRelation [_1#0, _2#1, _3#2]

== Physical Plan ==
     *(2) Project [id#6, name#7, score#32]
+- *(2) BroadcastHashJoin [id#6], [id#30], Inner, BuildRight
   :- LocalTableScan [id#6, name#7]
   +- BroadcastExchange HashedRelationBroadcastMode(List(cast(input[0, int, false] as bigint)))
      +- *(1) Project [_1#0 AS id#30, _3#2 AS score#32]
         +- *(1) Filter (_3#2 > 200)
            +- LocalTableScan [_1#0, _2#1, _3#2]

     */


    res1.show()

//    res1.explain(true)
  }

}
